package com.jqj.controller;

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import reactor.netty.http.client.HttpClientRequest;

import java.time.Duration;
import java.time.LocalTime;

/**
 * @Desc: 响应式流控制器
 * @Author: lijingzheng
 * @Date: 2025/1/8
 */
@RestController
public class ReactiveStreamController {

    @GetMapping("hello/{who}")
    public Flux<String> hello(@PathVariable String who) {

        int length = 10;
        String[] arr = new String[length];
        for (int i = 0; i < length; i++) {
            arr[i] = (who + "-" + i);
        }
        return Flux.just(arr)
                .map(w -> "<p style=\"color: red;\">Hello " + w + "!</p>");
    }

    @GetMapping("helloDelay/{who}")
    public Flux<String> helloDelay(@PathVariable String who) {

        int length = 10;
        String[] arr = new String[length];
        for (int i = 0; i < length; i++) {
            arr[i] = (who + "-" + i);
        }
        return Flux.just(arr)
                .map(w -> "<p style=\"color: red;\">Hello " + w + "!</p>")
                .delayElements(Duration.ofSeconds(1));
    }

    public static void main(String[] args) throws Exception {
        WebClient webClient = WebClient.builder().baseUrl("http://localhost:8888")
                .codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(1024 * 1024 * 10)).build();

//        Flux<String> flux = webClient.get().uri("/hello/LJZ").retrieve().bodyToFlux(String.class);
        Flux<String> flux = webClient.get().uri("/helloDelay/LJZ")
                .httpRequest(httpRequest -> {
                    HttpClientRequest reactorRequest = httpRequest.getNativeRequest();
                    reactorRequest.responseTimeout(Duration.ofSeconds(123));
                })
                // 设定期望的内容格式为 Server-Sent Events (SSE)
                .accept(MediaType.TEXT_EVENT_STREAM)
                .retrieve().bodyToFlux(String.class);

        flux.subscribeOn(Schedulers.boundedElastic())
                .doOnSubscribe(subscription -> println("Subscribed"))
                .doOnNext(item -> println("Received: " + item))
                .doOnError(error -> println("Error: " + error))
                .doOnComplete(() -> println("Completed"))
                .subscribe();

        System.in.read();
//        TimeUnit.SECONDS.sleep(100);
    }

    public static void println(String str) {
        System.out.printf("\033[34;1m%s ===>> %s ===>> %s\033[0m\n",
                LocalTime.now(), Thread.currentThread().getName(), str);
    }

}
